package com.pisolution.scheduler.runtime.master.dispatcher.executor;

import com.pisolution.scheduler.core.tuple.Tuple3;
import com.pisolution.scheduler.datasource.group.WorkerGroupMapper;
import com.pisolution.scheduler.remote.NettyRemoteClient;
import com.pisolution.scheduler.remote.command.Command;
import com.pisolution.scheduler.remote.command.CommandType;
import com.pisolution.scheduler.remote.utils.Host;
import com.pisolution.scheduler.runtime.master.processor.TaskAckProcessor;

import com.google.common.collect.Sets;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;

/**
 * 借助 {@link NettyRemoteClient} 实现的组件信息交互管理器.
 *
 * @author sora
 * @version 0.1.0
 */
@Service
public class NettyExecutorManager implements ExecutorManager<Boolean> {
    private static final Logger LOGGER = LoggerFactory.getLogger(NettyExecutorManager.class);

    // -------------------------------------------------------------------------------------------------

    /** 用于发送交互指令. */
    private final NettyRemoteClient client;

    /** 用于获取其他可用工作节点地址. */
    private final WorkerGroupMapper group;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    public NettyExecutorManager(final WorkerGroupMapper group) {
        this.client = new NettyRemoteClient();
        this.group = group;
    }

    // -------------------------------------------------------------------------------------------------
    //  生命周期
    // -------------------------------------------------------------------------------------------------

    /**
     * 用于依赖注入之后进行交互客户端指令解析器注册.
     *
     * <p><b>注意:</b></p>
     * <p>该方法会在构造函数与依赖注入之后自动调用 无需手动触发.</p>
     */
    @SuppressWarnings("unused")
    @PostConstruct
    public void configure() {
        this.client.register(CommandType.TASK_ACK_REQUEST, new TaskAckProcessor());
    }

    // -------------------------------------------------------------------------------------------------
    //  指令交互
    // -------------------------------------------------------------------------------------------------

    @Override
    public Boolean execute(Tuple3<Host, Command, String> context) throws Exception {
        // 1. 获取所有可用的工作节点.
        final Set<String> allServers = Sets.newHashSet(this.group.queryServerByName(context.field3).split(","));
        final Set<String> failedServers = new HashSet<>();

        Host host = context.field1;

        // 2. 如果尚未执行成功 则不断重试 直至无可用工作节点.
        while (true) {
            try {
                this.doExecute(host, context.field2);

                return true;
            } catch (Exception e) {
                LOGGER.error("Could not send the command {} to the worker server {}.", context.field2, context.field1, e);

                try {
                    // 3. 将失败的工作节点追加至失败的节点集合中.
                    failedServers.add(host.getAddress());

                    // 4. 如果尚有可用工作节点 则进行下一次尝试.
                    final Collection<String> remaining = Sets.difference(allServers, failedServers);
                    if (CollectionUtils.isNotEmpty(remaining)) {
                        host = Host.of(remaining.iterator().next());

                        LOGGER.error("Try to send the command {} to the worker server {}.", context.field2, host);
                        continue;
                    }

                    // 5. 无可用节点 判定为执行失败.
                    throw new Exception(String.format("Could not execute the command %s", context.field2));
                } catch (Exception ex) {
                    throw new Exception(ex.getMessage());
                }
            }
        }
    }

    @Override
    public void executeDirectly(Tuple3<Host, Command, String> context) throws Exception {
        this.doExecute(context.field1, context.field2);
    }

    /**
     * 向给定主机地址发送交互指令.
     *
     * @param host 接收指令的主机地址.
     * @param command 待发送的交互指令.
     *
     * @throws Exception 如果指令发送失败.
     */
    private void doExecute(final Host host, final Command command) throws Exception {
        // 1. 限定重试次数.
        int retryCount = 3;
        boolean isSuccess = false;

        do {
            try {
                this.client.send(host, command);
                isSuccess = true;
            } catch (Exception e) {
                LOGGER.error("Could not send the command {} to the worker server {}.", command, host, e);

                retryCount--;
                try {
                    Thread.sleep(100L);
                } catch (InterruptedException ignored) {
                    // 无需处理此异常.
                }
            }
        } while (retryCount >= 0 && !isSuccess);

        // 2. 如果执行失败 则反馈执行异常信息.
        if (!isSuccess) {
            throw new Exception(String.format("Could not send the command %s to the worker server %s.", command, host));
        }
    }
}
